package ithink.link.stack;

import ithink.link.base.message.LinkMessage;
import ithink.link.base.message.LinkMessageDecoder;
import ithink.link.base.message.LinkMessageEncoder;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/**
 * Created by zaqb on 17-9-11.
 */
public class LinkConnection extends SimpleChannelInboundHandler<LinkMessage> implements IConnection {
    private static final Logger LOGGER = LoggerFactory.getLogger(LinkConnection.class);
    private LinkStack stack = null;
    private State state;
    private String remoteIp = null;
    private Integer remotePort = null;
    private Integer localPort = null;
    private final Object stateSync = new Object();
    private ChannelHandlerContext ctx = null;
    private IConnectionListener listener;
    private final Object syncObj = new Object();
    private Object context;
    private LinkMessageDecoder msgDecoder;
    private LinkMessageEncoder msgEncoder;
    public LinkConnection(LinkStack stack,  LinkMessageDecoder msgDecoder, LinkMessageEncoder msgEncoder) {
        this.stack = stack;
        this.state = State.IDLE;
        this.msgDecoder = msgDecoder;
        this.msgEncoder = msgEncoder;
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        this.stack.exceptionCaught(new Exception(cause));
        disconnect(new Exception(cause));
        //TODO error conneciton
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        InetSocketAddress address = ((InetSocketAddress) ctx.channel().remoteAddress());
        this.remoteIp = address.getAddress().getHostAddress();
        this.remotePort = address.getPort();
        this.localPort = ((InetSocketAddress) ctx.channel().localAddress()).getPort();
        LOGGER.info("connection active local {}:remote {}", ctx.channel().localAddress().toString(), ctx.channel().remoteAddress().toString());
        boolean fire = false;
        synchronized (stateSync) {
            if (this.state == State.CONNECTING || this.state == State.IDLE) {
                this.state = State.CONNECTED;
                fire = true;
            } else if (state == State.DISCONNECTED) {
                ctx.disconnect();
                ctx.close();
                LOGGER.error("error state {} when active", state.toString());
                return;
            }
        }
        if (fire) {
            this.ctx = ctx;
            stack.registerConnection(this);
        }

    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        LOGGER.info("connection InActive local {}:remote {}", ctx.channel().localAddress().toString(), ctx.channel().remoteAddress().toString());
        boolean fire = false;
        synchronized (stateSync) {
            if (this.state == State.CONNECTED) {
                this.state = State.DISCONNECTED;
                fire = true;
            }
        }
        if (fire) {
            stack.unregisterConnection(this);
            onDisconnected(null);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

        if (evt instanceof IdleStateEvent) {
            IdleStateEvent e = (IdleStateEvent) evt;
            if (e.state() == IdleState.READER_IDLE) {
                    //time out
                LOGGER.info("close incoming connection {}:{} because read idle", remoteIp, remotePort);
                ctx.close();
            } else if (e.state() == IdleState.WRITER_IDLE) {
            }
        }
    }

    @Override
    public State getState() {
        return state;
    }

    @Override
    public String remoteIp() {
        return remoteIp;
    }

    @Override
    public int remotePort() {
        return remotePort;
    }

    @Override
    public Object getContext() {
        return context;
    }

    @Override
    public void setContext(Object context) {
        this.context = context;
    }

    public Integer getLocalPort() {
        return localPort;
    }

    @Override
    public void sendMessage(LinkMessage message) {
        if (state == State.CONNECTED) {
            LOGGER.info( "send message {} to {}", message.getText().toJSONString(), this.context );
            sendMessageInternal(message);
        } else {
            stack.onMessageSendFailed(this, message);
        }
    }

    @Override
    public void disconnect() {

    }

    @Override
    public void disconnect(Exception ex) {
        boolean fire = false;
        synchronized (stateSync) {
            if (state == State.CONNECTED) {
                //will fire inactive
                ctx.disconnect();
                ctx.close();
            } else if (state == State.CONNECTING || state == State.IDLE) {
                //only client type
                state = State.DISCONNECTED;
                fire = true;
            }
        }
        if (fire) {
            onDisconnected(ex);
        }
    }

    @Override
    public void setListener(IConnectionListener listener) {
        synchronized (syncObj) {
            this.listener = listener;
        }
    }




    @Override
    protected void messageReceived(ChannelHandlerContext channelHandlerContext, LinkMessage linkMessage) throws Exception {
        stack.onMessageReceived(this, linkMessage);
    }


    private void onDisconnected(Exception ex) {
        LOGGER.info("connection {}:{} disconnected", remoteIp, remotePort);
        if (ex != null) {
            LOGGER.error("because {}",ex);
        }
        synchronized (syncObj) {
            if(this.listener != null) {
                this.listener.onDisconnected(ex);
            }
        }
    }

    private void sendMessageInternal(final LinkMessage message) {
        LinkConnection self = this;
        ctx.writeAndFlush(message).addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (!future.isSuccess()) {
                    if (future.cause() != null) {
                        LOGGER.warn("send message error,{},{}", message, future.cause());
                    } else {
                        LOGGER.warn("send message error,{}", message);
                    }
                    stack.onMessageSendFailed(self, message);
                }
            }
        });
    }

    public void setFixedKey(String key) {
        this.msgDecoder.setFixedKey( key );
        this.msgEncoder.setFixedKey( key );
    }

}
